package com.killer.mq;


import com.alibaba.fastjson.JSON;
import com.killer.es.mapper.EsoOrderRepository;
import com.killer.es.entity.EsOrder;
import com.killer.pojo.entity.BOrder;
import com.killer.pojo.entity.CanalBean;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;
import java.util.Optional;

/***
 * 同步mysql和es
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = Constant.updateEs,consumerGroup = "canalListener",consumeMode = ConsumeMode.ORDERLY
, messageModel = MessageModel.CLUSTERING,
        consumeThreadMax = 1)
public class CanalListenerConsumer implements RocketMQListener<String> {

    @Autowired
    private EsoOrderRepository esoOrderRepository;

    @Override
    public void onMessage(String msg) {
        CanalBean canalBean = JSON.parseObject(msg,CanalBean.class);
        String table = canalBean.getTable();
        System.out.println(table);
        String type = canalBean.getType();
        System.out.println(type);
        List<BOrder> orders = canalBean.getData();
        orders.stream().forEach(item -> {
            System.out.println(item.toString());
            EsOrder order = new EsOrder();
            if("UPDATE".equals(type) && isTableExist(table)){
                Optional<EsOrder> order_item = esoOrderRepository.findById(item.getOrderId());
                if(order_item.isPresent()){
                    //更新
                    EsOrder esOrder = order_item.get();
                    BeanUtils.copyProperties(item,esOrder);
                    esoOrderRepository.save(esOrder);
                }else {
                   //执行添加操作
                    BeanUtils.copyProperties(item,order);
                    esoOrderRepository.save(order);
                }
            }else if("INSERT".equals(type) && isTableExist(table)){
                //执行插入操作
                BeanUtils.copyProperties(item,order);
                esoOrderRepository.save(order);
            }else {
                //执行删除操作
                BeanUtils.copyProperties(item,order);
                esoOrderRepository.delete(order);
            }
        });
    }

    protected boolean isTableExist(String tableName){
        List<String> list = Arrays.asList("b_order0","b_order1");
        return list.contains(tableName);
    }
}
